package com.aispeech.redis.test;

import io.lettuce.core.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class StreamConsumer {
    public static void main(String[] args) {
        String consumer_name = "consumer";

        RedisClient redisClient = RedisClient.create(RedisConf.CONN_STRING); // change to reflect your environment
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisCommands<String, String> syncCommands = connection.sync();

        try {
            syncCommands.xgroupCreate(XReadArgs.StreamOffset.from(RedisConf.KEY, "0-0"), RedisConf.GROUP);
        } catch (RedisBusyException redisBusyException) {
            log.warn("Group '{}' already exists", RedisConf.GROUP);
        }

        log.info("[ {} ] Waiting for new messages", consumer_name);

        while (true) {
            List<StreamMessage<String, String>> messages = syncCommands.xreadgroup(
                    Consumer.from(RedisConf.GROUP, consumer_name),
                    XReadArgs.StreamOffset.lastConsumed(RedisConf.KEY)
            );

            if (!messages.isEmpty()) {
                log.info("[ {} ] recv message...", consumer_name);

                for (StreamMessage<String, String> message : messages) {
                    String msgId = message.getId();
                    log.info("\tid: {}, message: {}", msgId, message.getBody());
                    syncCommands.xack(RedisConf.KEY, RedisConf.GROUP, msgId);
                }
            }
        }
    }
}
